package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-14 13:36
 */
public class Dwd_13_ToolCouponPay extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_13_ToolCouponPay().init(
                7007,
                3,
                "Dwd_13_ToolCouponPay"
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.读取ods层的数据
        readOdsDataFromKafka(tableEnv, "Dwd_13_ToolCouponPay");

        // 2.过滤出优惠券支付的数据
        Table couponUse = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['coupon_id'] coupon_id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['order_id'] order_id, " +
                "    date_format(`data`['used_time'], 'yyyy-MM-dd') date_id, " +
                "    `data`['used_time'] payment_time, " +
                "    `ts` " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'coupon_use' " +
                "    and `type` = 'update' " +
                "    and `data`['used_time'] is not null"
        );
        // 3.创建kafka-sink动态表
        tableEnv.executeSql(
                "create table dwd_tool_coupon_pay ( " +
                "    id string, " +
                "    coupon_id string, " +
                "    user_id string, " +
                "    order_id string, " +
                "    date_id string, " +
                "    payment_time string, " +
                "    ts bigint " +
                ")" + SqlUtil.getKafkaSinkDDL(Constants.TOPIC_DWD_TOOL_COUPON_PAY)
        );

        // 4.处理后的数据写到动态表中
        couponUse.executeInsert("dwd_tool_coupon_pay");
    }
}
